package indi.zhifa.study2025.flinkcdc.elsearch;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.UpdateRequest;
import co.elastic.clients.util.ContentType;
import com.alibaba.fastjson2.JSON;
import indi.zhifa.study2025.commondata.config.ElasticSearchClientProvider;
import indi.zhifa.study2025.flinkcdc.elsearch.command.ElCommand;
import indi.zhifa.study2025.flinkcdc.redis.RedisConfig;
import indi.zhifa.study2025.flinkcdc.redis.SpringDataRedisSink;
import indi.zhifa.study2025.flinkcdc.redis.command.RedisSinkCommand;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;

public class ElSearchSink<T> implements Sink<ElCommand<T>> {
    @Override
    public SinkWriter<ElCommand<T>> createWriter(InitContext context) throws IOException {
        return null;
    }

    @Override
    public SinkWriter<ElCommand<T>> createWriter(WriterInitContext context){
        return new ElSearchSink.ElSearchSinkWriter();
    }

    class ElSearchSinkWriter implements SinkWriter<ElCommand<T>> {

        @Override
        public void write(ElCommand<T> pCmd, Context context) throws IOException, InterruptedException {
            ElasticSearchClientProvider elasticSearchClientProvider = ElasticSearchClientProvider.getInstance();
            ElasticsearchClient elClient =  elasticSearchClientProvider.get();
            String index = pCmd.getIndex();
            String id = pCmd.getId();
            T entity = pCmd.getEntity();

            switch (pCmd.getCommand()){
                case CREATE,UPDATE -> {
                    elClient.index(i->i.index(index).id(id).document(entity));
                }
                case DELETE -> {
                    elClient.delete(d->d.index(index).id(id));
                }
            }
        }

        @Override
        public void flush(boolean endOfInput) throws IOException, InterruptedException {

        }

        @Override
        public void close() throws Exception {

        }
    }
}
